package main

import (
	"fmt"
	"github.com/aliyunmq/mq-http-go-sdk"
	"github.com/gogap/errors"
	"github.com/shima-park/agollo"
	"github.com/tidwall/gjson"
	"strings"
	"time"
)

type Message struct {
	CorpId    string `json:"CorpId"`
	SuiteId   string `json:"SuiteId"`
	Type      string `json:"Type"`
	Data      string `json:"Data"`
	Timestamp string `json:"Timestamp"`
}
type MessageData struct {
	ChgType        string `json:"ChgType"`
	UserId         string `json:"UserId"`
	ExternalUserId string `json:"ExternalUserId"`
	State          string `json:"State"`
	WelcomeCode    string `json:"WelcomeCode"`
}

func main() {
	// jsonData := `{"SuiteId":"ww858bec1e6067096d","CorpId":"wwb0761f38f2ff7fe4","Type":"group_updated","Data":"{\"Addmember\":[{\"MemberType\":1,\"Userid\":\"LiuChangTao\"}],\"Removemember\":[],\"ChatId\":\"wrTcD3CgAACkDcbHX1_DGQn3ky_FdEOw\"}","Timestamp":"2021/06/29 11:34:01"}`
	// fmt.Println("Addmember：",gjson.Get(gjson.Parse(gjson.Get(jsonData, "Data").String()).String(),"Addmember").Array())

	a, err := agollo.New("http://apollo.config.beta", "BetaGroupOperateXxlJob", agollo.AutoFetchOnCacheMiss())
	if err != nil {
		fmt.Println("链接apollo出错")
		panic(err)
	}

	app, api := a.GetNameSpace("application"), a.GetNameSpace("webapi")
	fmt.Println("app:",app,"api:",api)

	for key, value := range app {
		fmt.Println("健: ", key, "  值: ", value)
	}
	for key, value := range api {
		fmt.Println("健: ", key, "  值: ", value)
	}

	// 设置HTTP协议客户端接入点，进入消息队列RocketMQ版控制台实例详情页面的接入点区域查看。
	if app["Aliyun.MQ.Ons.EndPoint"]==nil {
		fmt.Println("获取Aliyun.MQ.Ons.EndPoint出错")
	}
	endpoint := app["Aliyun.MQ.Ons.EndPoint"].(string)
	// AccessKey ID阿里云身份验证，在阿里云RAM控制台创建。
	accessKey := app["Aliyun.MQ.Ons.AccessKey"].(string)
	// AccessKey Secret阿里云身份验证，在阿里云RAM控制台创建。
	secretKey := app["Aliyun.MQ.Ons.SecretKey"].(string)
	// 消息所属的Topic，在消息队列RocketMQ版控制台创建。
	//不同消息类型的Topic不能混用，例如普通消息的Topic只能用于收发普通消息，不能用于收发其他类型的消息。
	topic := app["Aliyun.MQ.Ons.Topic"].(string)
	// Topic所属的实例ID，在消息队列RocketMQ版控制台创建。
	// 若实例有命名空间，则实例ID必须传入；若实例无命名空间，则实例ID传入null空值或字符串空值。实例的命名空间可以在消息队列RocketMQ版控制台的实例详情页面查看。
	instanceId := app["Aliyun.MQ.Ons.InstanceId"].(string)
	// 您在控制台创建的Group ID。
	groupId := app["Aliyun.MQ.Ons.Group.Event"].(string)
	msgTags := app["Aliyun.MQ.Ons.Tag.Event"].(string)

	client := mq_http_sdk.NewAliyunMQClient(endpoint, accessKey, secretKey, "")

	mqConsumer := client.GetConsumer(instanceId, topic, groupId, msgTags)
	//fmt.Println(mqConsumer)
	for {
		endChan := make(chan int)
		respChan := make(chan mq_http_sdk.ConsumeMessageResponse)
		errChan := make(chan error)
		go func() {
			select {
			case resp := <-respChan:
				{
					// 处理业务逻辑。
					var handles []string
					fmt.Printf("Consume %d messages---->\n", len(resp.Messages))
					for _, v := range resp.Messages {
						handles = append(handles, v.ReceiptHandle)
						fmt.Printf("\tMessageID: %s, PublishTime: %d, MessageTag: %s\n"+
							"\tConsumedTimes: %d, FirstConsumeTime: %d, NextConsumeTime: %d\n"+
							"\tBody: %s\n"+
							"\tProps: %s\n",
							v.MessageId, v.PublishTime, v.MessageTag, v.ConsumedTimes,
							v.FirstConsumeTime, v.NextConsumeTime, v.MessageBody, v.Properties)

						businessType := gjson.Get(v.MessageBody, "Type")
						jsonData:=gjson.Parse(gjson.Get(v.MessageBody, "Data").String())
						fmt.Println("业务类型：", businessType)
						switch businessType.String() {
						case "group_update": //客户群相关业务
							addMember := gjson.Get(jsonData.String(), "Addmember").Array()
							if len(addMember)>0 {

							}
							break
						default://sync_one_extusers
							/*
							{"SuiteId":"wwf704d56d4467177c","CorpId":"ww24a91292508dddef","Type":"sync_one_extusers","Data":"{\"ChgType\":\"del_external_contact\",\"UserId\":\"syyu\",\"ExternalUserId\":\"woTcD3CgAAatck9brT_sIBCzu5eeofAA\",\"State\":\"\",\"WelcomeCode\":\"\"}","Timestamp":"2021/08/20 17:52:34"}
							*/
							break
						}

					}

					// NextConsumeTime前若不确认消息消费成功，则消息会被重复消费。
					// 消息句柄有时间戳，同一条消息每次消费拿到的都不一样。
					//ackerr := mqConsumer.AckMessage(handles)
					//if ackerr != nil {
					//	// 某些消息的句柄可能超时，会导致消息消费状态确认不成功。
					//	fmt.Println(ackerr)
					//	if errAckItems, ok := ackerr.(errors.ErrCode).Context()["Detail"].([]mq_http_sdk.ErrAckItem); ok {
					//		for _, errAckItem := range errAckItems {
					//			fmt.Printf("\t 出现异常:%s, 错误码:%s, 异常信息:%s\n",
					//				errAckItem.ErrorHandle, errAckItem.ErrorCode, errAckItem.ErrorMsg)
					//		}
					//	} else {
					//		fmt.Println("ack err =", ackerr)
					//	}
					//	time.Sleep(time.Duration(3) * time.Second)
					//} else {
					//	fmt.Printf("Ack ---->\n\t%s\n", handles)
					//}

					endChan <- 1
				}
			case err := <-errChan:
				{
					// Topic中没有消息可消费。
					if strings.Contains(err.(errors.ErrCode).Error(), "MessageNotExist") {
						fmt.Println("\n 暂未收到新消息，继续等待......!")
					} else {
						fmt.Println(err)
						time.Sleep(time.Duration(3) * time.Second)
					}
					endChan <- 1
				}
				//case <-time.After(35 * time.Second):
				//	{
				//		fmt.Println("Timeout of consumer message ??")
				//		endChan <- 1
				//	}
			}
		}()

		// 长轮询消费消息，网络超时时间默认为35s。
		// 长轮询表示如果Topic没有消息，则客户端请求会在服务端挂起3s，3s内如果有消息可以消费则立即返回响应。
		mqConsumer.ConsumeMessage(respChan, errChan,
			3, // 一次最多消费3条（最多可设置为16条）。
			3, // 长轮询时间3s（最多可设置为30s）。
		)
		<-endChan
	}
}
